package cn.itcast.function;

import cn.itcast.bean.SectorBean;
import cn.itcast.bean.StockBean;
import cn.itcast.util.DbUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Created by JiaYang Xu
 * Date 2021/7/6 16:33
 * Description 板块行情业务逻辑加工，秒级和分时行情共用此类
 */
public class SectorWindowFunction extends RichAllWindowFunction<StockBean, SectorBean, TimeWindow> {
    Map<String, List<Map<String, Object>>> sectorStockMap = null;
    Map<String, Map<String, Object>> sectorLastDayKlineMap = null;
    MapState<String, SectorBean> sectorMs = null;

    //3. 初始化基准价（1000）
    BigDecimal basePrice = new BigDecimal(1000l);

    @Override
    public void open(Configuration parameters) throws Exception {
        // 1. 初始化数据：板块对应关系、最近交易日日K（上一交易日）
        //板块对应关系
        String sql = "SELECT * FROM bdp_sector_stock";
        sectorStockMap = DbUtil.queryForGroup("sector_code", sql);
        //最近交易日日K（上一交易日）
        String sqlLast = "SELECT * FROM bdp_quot_sector_kline_day WHERE trade_date = (SELECT last_trade_date FROM tcc_date WHERE trade_date = CURDATE())";
        sectorLastDayKlineMap = DbUtil.query("sector_code", sqlLast);
        sectorMs = getRuntimeContext().getMapState(new MapStateDescriptor<String, SectorBean>("sectorMs", String.class, SectorBean.class));
    }

    @Override
    public void apply(TimeWindow window, Iterable<StockBean> values, Collector<SectorBean> out) throws Exception {
//        *
//         * 开发步骤
//         * 1.循环窗口内个股数据并缓存
//         * 2.轮询板块对应关系表下的个股数据、并获取指定板块下的个股
//         * 3.初始化全部数据
//         * 4.轮询板块下的个股数据，并获取板块名称、个股代码、个股流通股本和前一日板块总流通市值
//         * 5.计算：开盘价/收盘价累计流通市值、累计（交易量、交易金额）（注意：是个股累计值，获取缓存的个股数据）
//         *    累计个股开盘流通市值 = SUM(板块下的个股开盘价*个股流通股本)
//         *    累计个股收盘流通市值 = SUM(板块下的个股收盘价*个股流通股本)
//         * 6.判断是否是首日上市，并计算板块开盘价和收盘价
//         *    板块开盘价格 = 板块前收盘价*当前板块以开盘价计算的总流通市值/当前板块前一交易日板块总流通市值
//         *    板块当前价格 = 板块前收盘价*当前板块以收盘价计算的总流通市值/当前板块前一交易日板块总流通市值
//         * 7.初始化板块高低价
//         * 8.获取上一个窗口板块数据（高、低、成交量、成交金额）
//         * 9.计算最高价和最低价（前后窗口比较）
//         * 10.计算分时成交量和成交金额
//         * 11.开盘价与高低价比较
//         * 12.封装结果数据
//         * 13.缓存当前板块数据
       // 1.循环窗口内个股数据并缓存
        HashMap<String, StockBean> map = new HashMap<>();
        for (StockBean value : values) {
            map.put(value.getSecCode(),value);
        }
//        2.轮询板块对应关系表下的个股数据、并获取指定板块下的个股
        for (String sectorCode : sectorStockMap.keySet()) {
            List<Map<String, Object>> maps = sectorStockMap.get(sectorCode);
            //3.初始化全部数据
            BigDecimal preClosePrice = new BigDecimal(0);
            BigDecimal sectorHighPrice = new BigDecimal(0);
            BigDecimal sectorLowPrice = new BigDecimal(0);
            long sectorTradeVol = 0l; //分时成交量
            long sectorTradeAmt = 0l; //分时成交金额
            long sectorTradeVolDay = 0l; //日成交总量
            long sectorTradeAmtDay = 0l; //日成交总金额
            long eventTime = 0l; //取最新的个股时间
            long tradeTime = 0l; //取最新的个股时间
            String sectorName = null;
            BigDecimal sectorOpenPrie = new BigDecimal(0);
            BigDecimal sectorClosePrie = new BigDecimal(0);
            //4.轮询板块下的个股数据，并获取板块名称、个股代码、个股流通股本和前一日板块总流通市值
            //前一日板块总流通市值
            BigDecimal preSectorNegoCap = new BigDecimal(0);

            BigDecimal sumOpenNegoCap = new BigDecimal(0);//累计个股开盘流通市值
            BigDecimal sumCloseNegoCap = new BigDecimal(0);//累计个股收盘流通市值
            for (Map<String, Object> stringObjectMap : maps) {//板块下的个股列表数据
                sectorName = stringObjectMap.get("sector_name").toString();
                //个股证券代码
                String secCode = stringObjectMap.get("sec_code").toString();
                //个股流通股本
                BigDecimal negoCap = new BigDecimal(stringObjectMap.get("nego_cap").toString());
                preSectorNegoCap = new BigDecimal(stringObjectMap.get("pre_sector_nego_cap").toString());
                //5.计算：开盘价/收盘价累计流通市值、累计（交易量、交易金额）（注意：是个股累计值，获取缓存的个股数据）
                StockBean stockBean = map.get(secCode); //从实时流里面获取
                if (stockBean != null) { //如果不为空，说此个股及存在于实时流中，又存在于板块个股对应关系表中
                    eventTime = stockBean.getEventTime();
                    tradeTime = stockBean.getTradeTime();
                    //累计（交易量、交易金额）
                    sectorTradeVolDay += stockBean.getTradeVolDay();
                    sectorTradeAmtDay += stockBean.getTradeAmtDay();
                    //   累计个股开盘流通市值 = SUM(板块下的个股开盘价*个股流通股本)
                    //   累计个股收盘流通市值 = SUM(板块下的个股收盘价*个股流通股本)
                    BigDecimal openNegoCap = stockBean.getOpenPrice().multiply(negoCap).setScale(2, RoundingMode.HALF_UP);
                    BigDecimal closeNegoCap = stockBean.getClosePrice().multiply(negoCap).setScale(2, RoundingMode.HALF_UP);
                    sumOpenNegoCap = sumOpenNegoCap.add(openNegoCap);
                    sumCloseNegoCap = sumCloseNegoCap.add(closeNegoCap);
                }
            }

            //6.判断是否是首日上市，并计算板块开盘价和收盘价
            if(sectorLastDayKlineMap == null || sectorLastDayKlineMap.get(sectorCode) == null){  //首日
                preClosePrice = basePrice;
            }else{ //不是首日上市，需要从板块日K线表里面获取数据：前一日收盘价
                Map<String, Object> mapLast = sectorLastDayKlineMap.get(sectorCode);
                if(mapLast !=null){
                    preClosePrice = new BigDecimal(mapLast.get("close_price").toString());
                }
            }
            //板块开盘价格 = 板块前收盘价*当前板块以开盘价计算的总流通市值/当前板块前一交易日板块总流通市值
            sectorOpenPrie = preClosePrice.multiply(sumOpenNegoCap).setScale(2,BigDecimal.ROUND_HALF_UP).divide(preSectorNegoCap,2,RoundingMode.HALF_UP);
            //板块当前价格 = 板块前收盘价*当前板块以收盘价计算的总流通市值/当前板块前一交易日板块总流通市值
            sectorClosePrie = preClosePrice.multiply(sumCloseNegoCap).setScale(2,BigDecimal.ROUND_HALF_UP).divide(preSectorNegoCap,2,RoundingMode.HALF_UP);

            //7.初始化板块高低价
            sectorHighPrice = sectorClosePrie;
            sectorLowPrice = sectorClosePrie;

            //8.获取上一个窗口板块数据（高、低、成交量、成交金额）
            SectorBean sectorBeanLast = sectorMs.get(sectorCode);
            if(sectorBeanLast !=null){
                BigDecimal highPriceLast = sectorBeanLast.getHighPrice();
                BigDecimal lowPriceLast = sectorBeanLast.getLowPrice();
                Long tradeVolDayLast = sectorBeanLast.getTradeVolDay();
                Long tradeAmtDayLast = sectorBeanLast.getTradeAmtDay();
                //9.计算最高价和最低价（前后窗口比较）
                if(sectorHighPrice.compareTo(highPriceLast) == -1){
                    sectorHighPrice = highPriceLast;
                }
                if(sectorLowPrice.compareTo(lowPriceLast)  == 1){
                    sectorLowPrice = lowPriceLast;
                }
                //10.计算分时成交量和成交金额
                sectorTradeVol = sectorTradeVolDay - tradeVolDayLast;
                sectorTradeAmt = sectorTradeAmtDay - tradeAmtDayLast;
            }

            //11.开盘价与高低价比较
            if(sectorHighPrice.compareTo(sectorOpenPrie) == -1){
                sectorHighPrice = sectorOpenPrie;
            }
            if(sectorLowPrice.compareTo(sectorOpenPrie) == 1){
                sectorLowPrice = sectorOpenPrie;
            }

            //12.封装结果数据
            SectorBean sectorBean = new SectorBean();
            //eventTime、sectorCode、sectorName、preClosePrice、openPrice、highPrice、lowPrice、closePrice、
            //tradeVol、tradeAmt、tradeVolDay、tradeAmtDay、tradeTime
            sectorBean.setEventTime(eventTime);
            sectorBean.setSectorCode(sectorCode);
            sectorBean.setSectorName(sectorName);
            sectorBean.setPreClosePrice(preClosePrice);
            sectorBean.setOpenPrice(sectorOpenPrie);
            sectorBean.setHighPrice(sectorHighPrice);
            sectorBean.setLowPrice(sectorLowPrice);
            sectorBean.setClosePrice(sectorClosePrie);
            sectorBean.setTradeVol(sectorTradeVol);
            sectorBean.setTradeAmt(sectorTradeAmt);
            sectorBean.setTradeVolDay(sectorTradeVolDay);
            sectorBean.setTradeAmtDay(sectorTradeAmtDay);
            sectorBean.setTradeTime(tradeTime);
            out.collect(sectorBean);
            //13.缓存当前板块数据
            sectorMs.put(sectorCode,sectorBean);
            }
        }
    }

